package com.junbao.message.application.service.impl;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.commom.util.DateFormatUtil;
import com.junbao.message.application.service.MessageService;
import com.junbao.message.application.service.TaskService;
import com.junbao.message.domain.enums.ProducerName;
import com.junbao.message.domain.model.NotifyMessage;
import com.junbao.message.infrastructure.core.RequestDto;
import com.junbao.message.infrastructure.core.dispatcher.MessageConfirmDisruptor;
import com.junbao.message.infrastructure.core.dispatcher.MessageRetrySendDisruptor;
import com.junbao.message.infrastructure.page.DataQuereyCondition;
import com.junbao.message.infrastructure.repository.TimingTaskDao;

@SuppressWarnings("all")
public class TaskServiceImpl implements TaskService{

	private static final Logger LOGGER = LoggerFactory.getLogger(TaskServiceImpl.class);
	
	private static final String SWITCH_OFF = "off";
	
	private static final String SWITCH_ON = "on";
	
	private TimingTaskDao timingTaskDao;
	private MessageService messageService;
	
	public void messageConfirmSendFlagListener() {
		try{
			Integer number = timingTaskDao.updateMessageConfirmSwitch(SWITCH_OFF, SWITCH_ON);
			if (number==1) {
				Date conditionTime = new Date();
				//加载需要进行消息确认的消息
				List<NotifyMessage> waitConfirmMessages = messageService.queryWaitConfirmMessage(conditionTime);
				if (waitConfirmMessages.size() > 0) {
					//将加载的消息的消息状态置为确认中
					messageService.updateWaitConfirmMessageStatusToConfriming(conditionTime);
					for (NotifyMessage notifyMessage : waitConfirmMessages) {
						notifyMessage.markMessageToConfirming();
						//将消息传给disruptor进行消息确认
						MessageConfirmDisruptor.instance().
						getMessageProducer(ProducerName.MESSAGE_CONFIRM_PRODUCER)
						.sendData(new RequestDto<NotifyMessage>(notifyMessage));
					}
				}
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.error("定时任务进行消息重试失败！", e);
		}finally{
			try {
				Thread.sleep(1000);
				timingTaskDao.updateMessageConfirmSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.error("定时任务进行消息重试失败！", e);
			}
		}
	}

	
	public void messageRetrySendListener() {
		try{
			Integer number = timingTaskDao.updateMessageRetryTaskSwitch(SWITCH_OFF, SWITCH_ON);
			if (number==1) {
				//加载需要进行重发的消息
				Date conditionTime = new Date();
				int count = messageService.updateWaitSendMessageStatusToSending(conditionTime);
				if (count > 0) {
					List<NotifyMessage> waitSendMessages = messageService.queryWaitSendMessage(conditionTime);
					for (NotifyMessage notifyMessage : waitSendMessages) {
						//将消息传给disruptor进行重发
						MessageRetrySendDisruptor.instance().
						getMessageProducer(ProducerName.MESSAGE_RETRY_SEND_PRODUCER)
						.sendData(new RequestDto<NotifyMessage>(notifyMessage));
					}
				}
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.error("定时任务进行消息重发失败！", e);
		}finally{
			try {
				Thread.sleep(1000);
				timingTaskDao.updateMessageRetryTaskSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.error("定时任务进行消息重发失败！", e);
			}
		}
	}

	
	public void messageStatusToDeadListener() {
		try{
			Integer number = timingTaskDao.updateMessageToDeadTaskSwitch(SWITCH_OFF, SWITCH_ON);
			if (number==1) {
				messageService.updateMessageToDead();
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.error("更新超过重试次数的消息为死亡消息失败！", e);
		}finally{
			try {
				Thread.sleep(1000);
				timingTaskDao.updateMessageToDeadTaskSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.error("更新超过重试次数的消息为死亡消息失败！", e);
			}
		}
	}

	
	public void messageStatusToWaitConfirmListener() {
		try{
			Integer number = timingTaskDao.updateMessageToWaitConfirmTaskSwitch(SWITCH_OFF, SWITCH_ON);
			if (number==1) {
				messageService.updateStatusToWaitConfirm();
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.error("更新待通知消息为待确认状态失败！", e);
		}finally{
			try {
				Thread.sleep(1000);
				timingTaskDao.updateMessageToWaitConfirmTaskSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.error("更新待通知消息为待确认状态失败！", e);
			}
		}
	}
	
	
	public void messageRemovalToHistoryListener() {
		final ExecutorService executorService = Executors.newFixedThreadPool(20);
		try{
			Integer number = timingTaskDao.updateMessageRemovalSwitch(SWITCH_OFF, SWITCH_ON);
			if (number==1) {
				//获取当前时间的24小时之前的时间
				String conditionTime = DateFormatUtil.beforeDay(1, DateFormatUtil.DATE_FORMAT);
				DataQuereyCondition condition = new DataQuereyCondition(DateFormatUtil.strFormatDate(conditionTime, DateFormatUtil.DATE_FORMAT));
				//查询待移除的总记录数
				condition.initTotalPage(messageService.queryWaitRemvalTotalCount(condition));
				//消息移除
				CountDownLatch countDownLatch = new CountDownLatch(condition.getTotalCount());
				messageRemovalToHistory(condition,executorService,countDownLatch);
				countDownLatch.await();
				//删除已经被移除的消息表中数据
				messageService.deleteRemovaledMessageFromMessage(condition);
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.error("更新待通知消息为待确认状态失败！", e);
		}finally{
			try {
				Thread.sleep(1000);
				timingTaskDao.updateMessageRemovalSwitch(SWITCH_ON, SWITCH_OFF);
				executorService.awaitTermination(1, TimeUnit.HOURS);
				executorService.shutdown();
			} catch (Exception e) {
				LOGGER.error("更新待通知消息为待确认状态失败！", e);
			}
		}
	}
	
	@Override
	public void messageStatusToWaitConfirmFromConfirmingUpdate() {
		try{
			int number = timingTaskDao.messageStatusToWaitConfirmFromConfirmingUpdateSwitch(SWITCH_OFF, SWITCH_ON);
			if (number == 1) {
				messageService.messageStatusToWaitConfirmFromConfirmingUpdate();
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.info("更新消息为待确认失败！",e);
		}finally{
			try {
				int number = timingTaskDao.messageStatusToWaitConfirmFromConfirmingUpdateSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.info("更新消息为待确认失败！",e);
			}
		}
	}
	
	@Override
	public void messageStatusToWaitSendFromSendingUpdate() {
		try{
			int number = timingTaskDao.messageStatusToWaitSendFromSendingUpdateSwitch(SWITCH_OFF, SWITCH_ON);
			if (number == 1) {
				messageService.messageStatusToWaitSendFromSendingUpdate();
			}else {
				return;
			}
		}catch(Exception e){
			LOGGER.info("更新消息为待确认失败！",e);
		}finally{
			try {
				int number = timingTaskDao.messageStatusToWaitSendFromSendingUpdateSwitch(SWITCH_ON, SWITCH_OFF);
			} catch (Exception e) {
				LOGGER.info("更新消息为待确认失败！",e);
			}
		}
	}
	
	/**
	 * 移除消息到历史表
	 * @param condition
	 */
	private void messageRemovalToHistory(DataQuereyCondition condition,final ExecutorService executorService,final CountDownLatch countDownLatch){
		final List<NotifyMessage> waitRemovalMessages = messageService.queryWaitRemovalMessages(condition);
		executorService.execute(new Runnable() {
			
			public void run() {
				messageService.addWaitMessageToHistory(waitRemovalMessages);
				countDownLatch.countDown();
			}
		});
		if (condition.nextExecut()) {
			messageRemovalToHistory(condition.nextPageCount(),executorService,countDownLatch);
		}
	}
	
	public void setTimingTaskDao(TimingTaskDao timingTaskDao) {
		this.timingTaskDao = timingTaskDao;
	}

	public void setMessageService(MessageService messageService) {
		this.messageService = messageService;
	}

}
